package org.code.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.code.common.Constant;
import org.code.util.ByteStringConverter;
import org.code.util.ThreadUtil;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;

/**
 * @author xiaojian
 * <p>
 * 批量发送消息
 */
public class Batch {
    static final String TOPIC = "batch-topic";

    @Test
    public void batchProducer() throws Exception {
        // 创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");
        // 设置NameServer地址
        producer.setNamesrvAddr(Constant.NAME_SERVER);
        // 启动生产者
        producer.start();

        // 消息列表
        List<Message> messages = new ArrayList<>();
        messages.add(new Message(TOPIC, ByteStringConverter.stringToByteArray("batch message 1")));
        messages.add(new Message(TOPIC, ByteStringConverter.stringToByteArray("batch message 2")));
        messages.add(new Message(TOPIC, ByteStringConverter.stringToByteArray("batch message 3")));
        // 发送消息
        SendResult send = producer.send(messages);
        System.out.println("发送消息状态: " + send.getSendStatus());

        // 关闭生产者
        producer.shutdown();
    }

    @Test
    public void batchConsumer() throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch-consumer-group");
        // 设置NameServer地址
        consumer.setNamesrvAddr(Constant.NAME_SERVER);
        // 订阅主题
        consumer.subscribe(TOPIC, "*");
        // 注册一个消费监听 MessageListenerConcurrently 是并发消费
        // 默认是20个线程一起消费，可以参看 consumer.setConsumeThreadMax()
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt : list) {
                    System.out.println("接收消息: " + ByteStringConverter.byteArrayToString(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        // 阻塞
        ThreadUtil.read();
    }
}
